package com.kancy.framework.job.starter.parser;

import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.api.script.ScriptJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.executor.handler.JobProperties;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.kancy.framework.job.starter.annotation.ElasticJobConfig;
import com.kancy.framework.job.starter.job.AbstractScriptJob;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;

/**
 * @author kancy
 */
@Slf4j
@Builder
public class JobConfigParser {

    private final static String PREFIX = "elastic.job.";

    private Environment environment;
    private ApplicationContext applicationContext;
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    public void init() throws BeansException {
        environment = applicationContext.getEnvironment();
        // 获取spring容器中所有标注@ElasticJobConfig的bean
        Map<String, Object> beanMap = applicationContext.getBeansWithAnnotation(ElasticJobConfig.class);
        for (Object confBean : beanMap.values()) {
            if (!ElasticJob.class.isInstance(confBean)){
                log.warn("{} class has @ElasticJobConfig , but it is not instanceof {}", confBean.getClass().getName(), ElasticJob.class.getName());
                continue;
            }

            Class<?> clz = confBean.getClass();
            ElasticJobConfig conf = clz.getAnnotation(ElasticJobConfig.class);
            String jobTypeName = getJobTypeName(confBean, conf);

            String jobClass = clz.getName();
            String jobName = StringUtils.hasText(conf.name())?conf.name():clz.getSimpleName();
            String cron = getEnvironmentStringValue(jobName, JobAttributeTag.CRON, conf.cron());
            String shardingItemParameters = getEnvironmentStringValue(jobName, JobAttributeTag.SHARDING_ITEM_PARAMETERS, conf.shardingItemParameters());
            String description = getEnvironmentStringValue(jobName, JobAttributeTag.DESCRIPTION, conf.description());
            String jobParameter = getEnvironmentStringValue(jobName, JobAttributeTag.JOB_PARAMETER, conf.jobParameter());
            String jobExceptionHandler = getEnvironmentStringValue(jobName, JobAttributeTag.JOB_EXCEPTION_HANDLER, conf.jobExceptionHandler());
            String executorServiceHandler = getEnvironmentStringValue(jobName, JobAttributeTag.EXECUTOR_SERVICE_HANDLER, conf.executorServiceHandler());

            String jobShardingStrategyClass = getEnvironmentStringValue(jobName, JobAttributeTag.JOB_SHARDING_STRATEGY_CLASS, conf.jobShardingStrategyClass());
            String eventTraceRdbDataSource = getEnvironmentStringValue(jobName, JobAttributeTag.EVENT_TRACE_RDB_DATA_SOURCE, conf.eventTraceRdbDataSource());
            String scriptCommandLine = getEnvironmentStringValue(jobName, JobAttributeTag.SCRIPT_COMMAND_LINE, conf.scriptCommandLine());

            boolean failover = getEnvironmentBooleanValue(jobName, JobAttributeTag.FAILOVER, conf.failover());
            boolean misfire = getEnvironmentBooleanValue(jobName, JobAttributeTag.MISFIRE, conf.misfire());
            boolean overwrite = getEnvironmentBooleanValue(jobName, JobAttributeTag.OVERWRITE, conf.overwrite());
            boolean disabled = getEnvironmentBooleanValue(jobName, JobAttributeTag.DISABLED, conf.disabled());
            boolean monitorExecution = getEnvironmentBooleanValue(jobName, JobAttributeTag.MONITOR_EXECUTION, conf.monitorExecution());
            boolean streamingProcess = getEnvironmentBooleanValue(jobName, JobAttributeTag.STREAMING_PROCESS, conf.streamingProcess());

            int shardingTotalCount = getEnvironmentIntValue(jobName, JobAttributeTag.SHARDING_TOTAL_COUNT, conf.shardingTotalCount());
            int monitorPort = getEnvironmentIntValue(jobName, JobAttributeTag.MONITOR_PORT, conf.monitorPort());
            int maxTimeDiffSeconds = getEnvironmentIntValue(jobName, JobAttributeTag.MAX_TIME_DIFF_SECONDS, conf.maxTimeDiffSeconds());
            int reconcileIntervalMinutes = getEnvironmentIntValue(jobName, JobAttributeTag.RECONCILE_INTERVAL_MINUTES, conf.reconcileIntervalMinutes());

            // 核心配置
            JobCoreConfiguration coreConfig =
                    JobCoreConfiguration.newBuilder(jobName, cron, shardingTotalCount)
                            .shardingItemParameters(shardingItemParameters)
                            .description(description)
                            .failover(failover)
                            .jobParameter(jobParameter)
                            .misfire(misfire)
                            .jobProperties(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), jobExceptionHandler)
                            .jobProperties(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(), executorServiceHandler)
                            .build();

            // 不同类型的任务配置处理
            LiteJobConfiguration jobConfig = null;
            JobTypeConfiguration typeConfig = null;
            if (jobTypeName.equalsIgnoreCase("SimpleJob")) {
                typeConfig = new SimpleJobConfiguration(coreConfig, jobClass);
            }

            if (jobTypeName.equalsIgnoreCase("DataflowJob")) {
                typeConfig = new DataflowJobConfiguration(coreConfig, jobClass, streamingProcess);
            }

            if (jobTypeName.equalsIgnoreCase("ScriptJob")) {
                String script = scriptCommandLine;
                // 配置的形式脚本为空时，才使用抽象方法提供的脚本
                if (script == null || script.trim().length() == 0){
                    if (confBean instanceof AbstractScriptJob){
                        script = ((AbstractScriptJob)confBean).scriptCommandLine();
                    }
                }
                if (script == null || script.trim().length() == 0){
                    log.warn("{} script job scriptCommandLine is empty.", jobName);
                }
                typeConfig = new ScriptJobConfiguration(coreConfig, script);
            }

            jobConfig = LiteJobConfiguration.newBuilder(typeConfig)
                    .overwrite(overwrite)
                    .disabled(disabled)
                    .monitorPort(monitorPort)
                    .monitorExecution(monitorExecution)
                    .maxTimeDiffSeconds(maxTimeDiffSeconds)
                    .jobShardingStrategyClass(jobShardingStrategyClass)
                    .reconcileIntervalMinutes(reconcileIntervalMinutes)
                    .build();

            List<BeanDefinition> elasticJobListeners = getTargetElasticJobListeners(conf);

            // 构建SpringJobScheduler对象来初始化任务
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
            factory.setScope(BeanDefinition.SCOPE_PROTOTYPE);
            if ("ScriptJob".equals(jobTypeName)) {
                factory.addConstructorArgValue(null);
            } else {
                factory.addConstructorArgValue(confBean);
            }
            factory.addConstructorArgValue(zookeeperRegistryCenter);
            factory.addConstructorArgValue(jobConfig);

            // 任务执行日志数据源，以名称获取
            if (StringUtils.hasText(eventTraceRdbDataSource)) {
                BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(JobEventRdbConfiguration.class);
                rdbFactory.addConstructorArgReference(eventTraceRdbDataSource);
                factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
            }

            factory.addConstructorArgValue(elasticJobListeners);
            DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
            String jobSchedulerBeanName = jobName + "SpringJobScheduler";
            defaultListableBeanFactory.registerBeanDefinition(jobSchedulerBeanName, factory.getBeanDefinition());
            SpringJobScheduler springJobScheduler = (SpringJobScheduler) applicationContext.getBean(jobSchedulerBeanName);
            springJobScheduler.init();
            log.info("[" + jobName + "]\t" + jobClass + "\t初始化成功");
        }
    }

    private String getJobTypeName(Object confBean, ElasticJobConfig conf) {
        if(StringUtils.hasText(conf.typeName())){
            return conf.typeName();
        }

        // 当前类实现的接口，优先级按顺序排序
        Class<?>[] interfaces = confBean.getClass().getInterfaces();
        if(interfaces != null && interfaces.length > 0){
            for (int i = 0; i < interfaces.length; i++) {
                if(interfaces[i] == SimpleJob.class){
                    return SimpleJob.class.getSimpleName();
                }
                if(interfaces[i] == DataflowJob.class){
                    return SimpleJob.class.getSimpleName();
                }
                if(interfaces[i] == ScriptJob.class){
                    return SimpleJob.class.getSimpleName();
                }
            }
        }
        // 当前的Job类没有接口时，看他是哪一种类型
        if (confBean instanceof SimpleJob){
            return SimpleJob.class.getSimpleName();
        }
        if (confBean instanceof DataflowJob){
            return DataflowJob.class.getSimpleName();
        }
        if (confBean instanceof ScriptJob){
            return ScriptJob.class.getSimpleName();
        }

        throw new RuntimeException(String.format("Not found %s Job TypeName.", confBean.getClass().getName()));
    }

    private List<BeanDefinition> getTargetElasticJobListeners(ElasticJobConfig conf) {
        List<BeanDefinition> result = new ManagedList<>(2);
        String listeners = getEnvironmentStringValue(conf.name(), JobAttributeTag.LISTENER, conf.listener());
        if (StringUtils.hasText(listeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(listeners);
            factory.setScope(BeanDefinition.SCOPE_PROTOTYPE);
            result.add(factory.getBeanDefinition());
        }

        long startedTimeoutMilliseconds = getEnvironmentLongValue(conf.name(), JobAttributeTag.DISTRIBUTED_LISTENER_STARTED_TIMEOUT_MILLISECONDS, conf.startedTimeoutMilliseconds());
        long completedTimeoutMilliseconds = getEnvironmentLongValue(conf.name(), JobAttributeTag.DISTRIBUTED_LISTENER_COMPLETED_TIMEOUT_MILLISECONDS, conf.completedTimeoutMilliseconds());

        String distributedListeners = getEnvironmentStringValue(conf.name(), JobAttributeTag.DISTRIBUTED_LISTENER, conf.distributedListener());
        if (StringUtils.hasText(distributedListeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(distributedListeners);
            factory.setScope(BeanDefinition.SCOPE_PROTOTYPE);
            factory.addConstructorArgValue(startedTimeoutMilliseconds);
            factory.addConstructorArgValue(completedTimeoutMilliseconds);
            result.add(factory.getBeanDefinition());
        }
        return result;
    }

    /**
     * 获取配置中的任务属性值，environment没有就用注解中的值
     *
     * @param jobName      任务名称
     * @param fieldName    属性名称
     * @param defaultValue 默认值
     */
    private String getEnvironmentStringValue(String jobName, String fieldName, String defaultValue) {
        String key = PREFIX + jobName + "." + fieldName;
        String value = environment.getProperty(key);
        if (StringUtils.hasText(value)) {
            return value;
        }
        return defaultValue;
    }

    private int getEnvironmentIntValue(String jobName, String fieldName, int defaultValue) {
        String key = PREFIX + jobName + "." + fieldName;
        String value = environment.getProperty(key);
        if (StringUtils.hasText(value)) {
            return Integer.valueOf(value);
        }
        return defaultValue;
    }

    private long getEnvironmentLongValue(String jobName, String fieldName, long defaultValue) {
        String key = PREFIX + jobName + "." + fieldName;
        String value = environment.getProperty(key);
        if (StringUtils.hasText(value)) {
            return Long.valueOf(value);
        }
        return defaultValue;
    }

    private boolean getEnvironmentBooleanValue(String jobName, String fieldName, boolean defaultValue) {
        String key = PREFIX + jobName + "." + fieldName;
        String value = environment.getProperty(key);
        if (StringUtils.hasText(value)) {
            return Boolean.valueOf(value);
        }
        return defaultValue;
    }
}
